#include "in.h"

int CloseAcceptedSession( struct MQAcceptedSession *p_accepted_session )
{
	INFOLOG( "close session[%p]#%d# client[%s:%d]" , p_accepted_session,p_accepted_session->accepted_sock , p_accepted_session->client_ip , p_accepted_session->client_port )
	
	if( p_accepted_session->comm_data )
	{
		DEBUGLOG( "FreeBuffer .comm_data" )
		FreeBuffer( p_accepted_session->comm_data );
	}
	
	if( p_accepted_session->p_subscribe_group )
	{
		if( p_accepted_session->p_subscribe_group->topic_fill_step > 0 )
		{
			DEBUGLOG( "Feedback subscribe_group[%s] topic_fill_step[%"PRIu64"]" , p_accepted_session->p_subscribe_group->subscribe_group_name , p_accepted_session->p_subscribe_group->topic_fill_step )
			p_accepted_session->p_subscribe_group->topic_fill_ptr -= p_accepted_session->p_subscribe_group->topic_fill_step ;
		}
	}
	
	if( ! list_empty( & (p_accepted_session->subscribe_group_subscribe_session_list_node) ) )
	{
		DEBUGLOG( "list_del .subscribe_group_subscribe_session_list_node" )
		list_del( & (p_accepted_session->subscribe_group_subscribe_session_list_node) );
		
		if( ! list_empty( & (p_accepted_session->p_subscribe_group->subscribe_session_list) ) )
		{
			DEBUGLOG( "Re FetchSubscribeGroupMessage subscribe_group[%s]" , p_accepted_session->p_subscribe_group->subscribe_group_name )
			FetchSubscribeGroupMessage( p_accepted_session->p_tcpdaemon_env , p_accepted_session->p_subscribe_group );
		}
		else
		{
			DEBUGLOG( "UnlinkSubscribeGroupNameTreeNode and free subscribe_group[%s]" , p_accepted_session->p_subscribe_group->subscribe_group_name )
			UnlinkSubscribeGroupNameTreeNode( p_accepted_session->p_subscribe_group->p_topic , p_accepted_session->p_subscribe_group );
			free( p_accepted_session->p_subscribe_group );
		}
	}
	
	if( ! list_empty( & (p_accepted_session->msg_file_subscribe_session_list_node) ) )
	{
		DEBUGLOG( "list_del .msg_file_subscribe_session_list_node" )
		list_del( & (p_accepted_session->msg_file_subscribe_session_list_node) );
	}
	
	if( ! list_empty( & (p_accepted_session->topic_file_subscribe_session_list_node) ) )
	{
		DEBUGLOG( "list_del .topic_file_subscribe_session_list_node" )
		list_del( & (p_accepted_session->topic_file_subscribe_session_list_node) );
	}
	
	DEBUGLOG( "free p_accepted_session[%p]" , p_accepted_session )
	free( p_accepted_session );
	
	return 0;
}

func_tcpmain tcpmain ;
int tcpmain( struct TcpdaemonServerEnvironment *p_env , int accepted_sock , void *p_addr )
{
	struct MQBroker			*broker = (struct MQBroker *)TDGetTcpmainParameter(p_env) ;
	struct MQAcceptedSession	*p_accepted_session = NULL ;
	struct MQStore			*p_store = NULL ;
	struct MQDelayMessage		*delay_msg = NULL ;
	
	int				nret = 0 ;
	
	switch( TDGetIoMultiplexEvent(p_env) )
	{
		case IOMP_ON_INIT_WORKER :
			
			p_store = NULL ;
			for( ; ; )
			{
				p_store = TravelStoreNameTreeNode( broker , p_store ) ;
				if( p_store == NULL )
					break;
				
				delay_msg = NULL ;
				for( ; ; )
				{
					delay_msg = TravelDelayMessageSendTimestampTreeNode( p_store , delay_msg ) ;
					if( delay_msg == NULL )
						break;
					
					DEBUGLOG( "TDAddCustemSessionToEventBus store[%s] delay_msg[%p]->delay_timer[%d]->delay_seconds[%"PRIu32"]" , p_store->store_name , delay_msg , delay_msg->delay_timer , delay_msg->delay_seconds )
					TDAddCustemSessionToEventBus( p_env , delay_msg->delay_timer , delay_msg );
				}
			}
			
			return TCPMAIN_RETURN_SILENCE;
			
		case IOMP_ON_ACCEPTING_SOCKET :
			
			if( broker->tcp_nodelay )
			{
				int	onoff = 1 ;
				setsockopt( accepted_sock , IPPROTO_TCP , TCP_NODELAY , (void*) & onoff , sizeof(int) );
				DEBUGLOG( "IOMP_ON_ACCEPTING_SOCKET - enable socket TCP_NODELAY" )
			}
			
			p_accepted_session = (struct MQAcceptedSession *)malloc( sizeof(struct MQAcceptedSession) ) ;
			if( p_accepted_session == NULL )
			{
				ERRORLOG( "IOMP_ON_ACCEPTING_SOCKET - alloc accepted session failed , errno[%d]" , errno )
				return TCPMAIN_RETURN_CLOSE;
			}
			else
			{
				DEBUGLOG( "malloc p_accepted_session[%p]" , p_accepted_session )
			}
			memset( p_accepted_session , 0x00 , sizeof(struct MQAcceptedSession) );
			
			p_accepted_session->accepted_sock = accepted_sock ;
			memcpy( & (p_accepted_session->accepted_address) , p_addr , sizeof(struct sockaddr) );
			strcpy( p_accepted_session->client_ip , inet_ntoa(p_accepted_session->accepted_address.sin_addr) );
			p_accepted_session->client_port = (int)ntohs( p_accepted_session->accepted_address.sin_port ) ;
			INFOLOG( "IOMP_ON_ACCEPTING_SOCKET - accept#%d# client[%s:%d]" , p_accepted_session->accepted_sock , p_accepted_session->client_ip , p_accepted_session->client_port )
			
			p_accepted_session->p_tcpdaemon_env = p_env ;
			
			p_accepted_session->comm_data = AllocBufferInitSize( 1024 ) ;
			if( p_accepted_session->comm_data == NULL )
			{
				ERRORLOG( "IOMP_ON_ACCEPTING_SOCKET - alloc comm data failed , errno[%d]" , errno )
				free( p_accepted_session );
				return TCPMAIN_RETURN_CLOSE;
			}
			
			p_accepted_session->broker = TDGetTcpmainParameter( p_env ) ;
			
			INIT_LIST_HEAD( & (p_accepted_session->subscribe_group_subscribe_session_list_node) );
			
			INIT_LIST_HEAD( & (p_accepted_session->msg_file_subscribe_session_list_node) );
			INIT_LIST_HEAD( & (p_accepted_session->topic_file_subscribe_session_list_node) );
			
			DEBUGLOG( "IOMP_ON_ACCEPTING_SOCKET - TDSetIoMultiplexDataPtr p_accepted_session[%p]" , p_accepted_session )
			TDSetIoMultiplexDataPtr( p_env , p_accepted_session );
			
			return (p_accepted_session->comm_status=TCPMAIN_RETURN_WAITINGFOR_RECEIVING);
			
		case IOMP_ON_RECEIVING_SOCKET :
			
			p_accepted_session = (struct MQAcceptedSession *) p_addr ;
			DEBUGLOG( "IOMP_ON_RECEIVING_SOCKET - accepted_session[%p]#%d#" , p_accepted_session,p_accepted_session->accepted_sock )
			
_GOTO_RECEIVE_AGAIN :
			DEBUGLOG( "IOMP_ON_RECEIVING_SOCKET - TDHBReceiveDataWithNonblock ... , long_pull_as_server_status[%"PRIu16"]" , p_accepted_session->long_pull_as_server_status )
			nret = TDHBReceiveDataWithNonblock( p_accepted_session->accepted_sock , MQ_COMM_PREHEAD_LEN , GetAppendBufferBase(p_accepted_session->comm_data) , GetAppendBufferSize(p_accepted_session->comm_data) , & (p_accepted_session->tdhb) ) ;
			INFOLOG( "IOMP_ON_RECEIVING_SOCKET - TDHBReceiveDataWithNonblock return [%d]" , nret )
			if( nret == TDHB_ERROR_SOCKET_CLOSED )
			{
				if( p_accepted_session->tdhb.head_remain_len == MQ_COMM_PREHEAD_LEN )
				{
					DEBUGLOG( "IOMP_ON_RECEIVING_SOCKET - return [%d]TCPMAIN_RETURN_CLOSE" , TCPMAIN_RETURN_CLOSE )
					return (p_accepted_session->comm_status=TCPMAIN_RETURN_CLOSE);
				}
				else
				{
					ERRORLOG( "IOMP_ON_RECEIVING_SOCKET - return [%d]TCPMAIN_RETURN_CLOSE" , TCPMAIN_RETURN_CLOSE )
					return (p_accepted_session->comm_status=TCPMAIN_RETURN_CLOSE);
				}
			}
			else if( nret == TCPMAIN_RETURN_WAITINGFOR_NEXT )
			{
				DEBUGLOG( "IOMP_ON_RECEIVING_SOCKET - return [%d]TCPMAIN_RETURN_WAITINGFOR_NEXT" , nret )
				return (p_accepted_session->comm_status=nret);
			}
			else if( nret == TCPMAIN_RETURN_WAITINGFOR_SENDING )
			{
				DEBUGLOG( "before change long_pull_as_server_status - p_accepted_session[%p] ->accepted_sock[%d] ->long_pull_as_server_status[%"PRIu16"]" , p_accepted_session , p_accepted_session->accepted_sock , p_accepted_session->long_pull_as_server_status )
				if( p_accepted_session->long_pull_as_server_status == MQ_LONG_PULL_AS_SERVER_STATUS_LAST_RECV_BEFORE_SILENT )
					p_accepted_session->long_pull_as_server_status = MQ_LONG_PULL_AS_SERVER_STATUS_SILENT ;
				DEBUGLOG( "after change long_pull_as_server_status - p_accepted_session[%p] ->accepted_sock[%d] ->long_pull_as_server_status[%"PRIu16"]" , p_accepted_session , p_accepted_session->accepted_sock , p_accepted_session->long_pull_as_server_status )
				
				OffsetAppendBufferFillPtr( p_accepted_session->comm_data , p_accepted_session->tdhb.body_len );
				
				p_accepted_session->tcpdaemon_session = TDGetIoMultiplexCurrentSession( p_env ) ;
				
				nret = DispatchProcess( p_env , p_accepted_session ) ;
				if( nret == TCPMAIN_RETURN_CLOSE )
				{
					ERRORLOG( "IOMP_ON_RECEIVING_SOCKET - DispatchProcess return [%d]" , nret )
					ERRORLOG( "IOMP_ON_RECEIVING_SOCKET - return [%d]TCPMAIN_RETURN_CLOSE" , TCPMAIN_RETURN_CLOSE )
					return (p_accepted_session->comm_status=TCPMAIN_RETURN_CLOSE);
				}
				else
				{
					INFOLOG( "IOMP_ON_RECEIVING_SOCKET - DispatchProcess return [%d]" )
					DEBUGLOG( "IOMP_ON_RECEIVING_SOCKET - return [%d]%s" , nret , g_tcpdaemon_tcpmain_return_value_desc[nret] )
					return (p_accepted_session->comm_status=nret);
				}
			}
			else if( nret == TDHB_ERROR_RECV_BUFFER_OVERFLOW )
			{
				nret = ResizeAppendBuffer( p_accepted_session->comm_data , p_accepted_session->tdhb.body_len+1 ) ;
				if( nret )
				{
					ERRORLOG( "IOMP_ON_RECEIVING_SOCKET - ResizeAppendBuffer failed , errno[%d]" , errno )
					ERRORLOG( "IOMP_ON_RECEIVING_SOCKET - return [%d]TCPMAIN_RETURN_CLOSE" , TCPMAIN_RETURN_CLOSE )
					return (p_accepted_session->comm_status=TCPMAIN_RETURN_CLOSE);
				}
				else
				{
					DEBUGLOG( "IOMP_ON_RECEIVING_SOCKET - ResizeAppendBuffer ok , new size[%d]bytes" , p_accepted_session->tdhb.body_len+1 )
					goto _GOTO_RECEIVE_AGAIN;
				}
			}
			else
			{
				ERRORLOG( "IOMP_ON_RECEIVING_SOCKET - return [%d]" , nret )
				return (p_accepted_session->comm_status=nret);
			}
			
			break;
			
		case IOMP_ON_SENDING_SOCKET :
			
			p_accepted_session = (struct MQAcceptedSession *) p_addr ;
			DEBUGLOG( "IOMP_ON_SENDING_SOCKET - accepted_session[%p]#%d#" , p_accepted_session,p_accepted_session->accepted_sock )
			
			DEBUGLOG( "IOMP_ON_SENDING_SOCKET - TDHBSendDataWithNonblock ... , long_pull_as_server_status[%"PRIu16"]" , p_accepted_session->long_pull_as_server_status )
			nret = TDHBSendDataWithNonblock( p_accepted_session->accepted_sock , MQ_COMM_PREHEAD_LEN , GetAppendBufferBase(p_accepted_session->comm_data) , GetAppendBufferLength(p_accepted_session->comm_data) , & (p_accepted_session->tdhb) ) ;
			if( nret == TCPMAIN_RETURN_WAITINGFOR_NEXT )
			{
				DEBUGLOG( "IOMP_ON_RECEIVING_SOCKET - return [%d]TCPMAIN_RETURN_WAITINGFOR_NEXT" , nret )
				return (p_accepted_session->comm_status=nret);
			}
			else if( nret == TCPMAIN_RETURN_WAITINGFOR_RECEIVING )
			{
				DEBUGLOG( "before change long_pull_as_server_status - p_accepted_session[%p] ->accepted_sock[%d] ->long_pull_as_server_status[%"PRIu16"]" , p_accepted_session , p_accepted_session->accepted_sock , p_accepted_session->long_pull_as_server_status )
				if( p_accepted_session->long_pull_as_server_status == MQ_LONG_PULL_AS_SERVER_STATUS_LAST_SEND_AFTER_SILENT )
					p_accepted_session->long_pull_as_server_status = MQ_LONG_PULL_AS_SERVER_STATUS_SILENT ;
				else if( p_accepted_session->long_pull_as_server_status == MQ_LONG_PULL_AS_SERVER_STATUS_FIRST_SEND_BEFORE_SILENT )
					p_accepted_session->long_pull_as_server_status = MQ_LONG_PULL_AS_SERVER_STATUS_LAST_RECV_BEFORE_SILENT ;
				DEBUGLOG( "after change long_pull_as_server_status - p_accepted_session[%p] ->accepted_sock[%d] ->long_pull_as_server_status[%"PRIu16"]" , p_accepted_session , p_accepted_session->accepted_sock , p_accepted_session->long_pull_as_server_status )
				
				DEBUGLOG( "IOMP_ON_RECEIVING_SOCKET - return [%d]TCPMAIN_RETURN_WAITINGFOR_RECEIVING" , nret )
				return (p_accepted_session->comm_status=TCPMAIN_RETURN_WAITINGFOR_RECEIVING);
			}
			else
			{
				ERRORLOG( "IOMP_ON_SENDING_SOCKET - return [%d]" , nret )
				return (p_accepted_session->comm_status=nret);
			}
			
			break;
			
		case IOMP_ON_CLOSING_SOCKET :
			
			p_accepted_session = (struct MQAcceptedSession *) p_addr ;
			DEBUGLOG( "IOMP_ON_CLOSING_SOCKET - accepted_session[%p]#%d#" , p_accepted_session,p_accepted_session->accepted_sock )
			
			CloseAcceptedSession( p_accepted_session );
			
			return TCPMAIN_RETURN_SILENCE;
			
		case IOMP_ON_REBOOT_WORKER :
			
			UnloadAllMessageStores( broker );
			
			nret = LoadAllMessageStores( broker ) ;
			if( nret )
			{
				ERRORLOG( "LoadAllMessageStore failed[%d]" , nret )
				return (p_accepted_session->comm_status=TCPMAIN_RETURN_ERROR);
			}
			else
			{
				INFOLOG( "Load all message p_store ok" )
			}
			
			return TCPMAIN_RETURN_SILENCE;
			
		case IOMP_ON_IDLE_PROCESS :
			
			p_accepted_session = (struct MQAcceptedSession *) p_addr ;
			DEBUGLOG( "IOMP_ON_IDLE_PROCESS - accepted_session[%p]#%d#" , p_accepted_session,p_accepted_session->accepted_sock )
			
			nret = app_Heartbeat( p_env , p_accepted_session ) ;
			if( nret < 0 )
			{
				ERRORLOG( "app_Heartbeat accepted_session[%p] return [%d]" , p_accepted_session , nret )
				return TCPMAIN_RETURN_CLOSE;
			}
			else
			{
				DEBUGLOG( "app_Heartbeat accepted_session[%p] return [%d]" , p_accepted_session , nret )
			}
			
			return nret;
			
		case IOMP_ON_CUSTOMEVENT1_PROCESS :
			
			delay_msg = (struct MQDelayMessage *) p_addr ;
			DEBUGLOG( "IOMP_ON_CUSTOMEVENT1_PROCESS - delay_session[%p] delay_timer#%d#" , delay_msg,delay_msg->delay_timer )
			
			nret = app_AddDelayMessage( p_env , delay_msg ) ;
			if( nret )
			{
				ERRORLOG( "app_AddDelayMessage delay_msg[%p] return [%d]" , delay_msg , nret )
			}
			else
			{
				DEBUGLOG( "app_AddDelayMessage delay_msg[%p] return [%d]" , delay_msg , nret )
			}
			
			TDRemoveCustemSessionFromEventBus( p_env , delay_msg->delay_timer );
			close( delay_msg->delay_timer );
			
			return TCPMAIN_RETURN_SILENCE;
			
		default :
			return TCPMAIN_RETURN_DEFAULT;
	}
}

